package com.huan.source

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011

object KafkaSource {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val topic = "sensor"
    val props = new Properties()

    props.setProperty("bootstrap.servers","Bigdata:9092")
    props.setProperty("group.id","consumer-group")

    val inputStream = env.addSource(new FlinkKafkaConsumer011[String](topic,new SimpleStringSchema(),props))

    inputStream.print()

    env.execute("KafkaSource")



  }

}
